package com.atguigu.dga.governance.service.impl;

import com.atguigu.dga.common.utils.SpringBeanProvider;
import com.atguigu.dga.ds.bean.TDsTaskDefinition;
import com.atguigu.dga.ds.bean.TDsTaskInstance;
import com.atguigu.dga.ds.service.TDsTaskDefinitionService;
import com.atguigu.dga.ds.service.TDsTaskInstanceService;
import com.atguigu.dga.governance.assessor.Assessor;
import com.atguigu.dga.governance.bean.AssessParam;
import com.atguigu.dga.governance.bean.GovernanceAssessDetail;
import com.atguigu.dga.governance.bean.GovernanceMetric;
import com.atguigu.dga.governance.mapper.GovernanceAssessDetailMapper;
import com.atguigu.dga.governance.service.GovernanceAssessDetailService;
import com.atguigu.dga.governance.service.GovernanceMetricService;
import com.atguigu.dga.meta.bean.TableMetaInfo;
import com.atguigu.dga.meta.service.TableMetaInfoService;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import org.apache.hadoop.yarn.webapp.hamlet2.Hamlet;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import com.baomidou.dynamic.datasource.annotation.DS;

import javax.swing.*;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * <p>
 * 治理考评结果明细 服务实现类
 * </p>
 *
 * @author zhangchen
 * @since 2024-07-27
 */
@Service
@DS("dga")
public class GovernanceAssessDetailServiceImpl extends ServiceImpl<GovernanceAssessDetailMapper, GovernanceAssessDetail> implements GovernanceAssessDetailService {

    @Autowired
    TableMetaInfoService tableMetaInfoService;

    @Autowired
    GovernanceMetricService governanceMetricService;

    @Autowired
    TDsTaskDefinitionService tDsTaskDefinitionService;

    @Autowired
    TDsTaskInstanceService tDsTaskInstanceService;

    ThreadPoolExecutor threadPoolExecutor=  new ThreadPoolExecutor(10,20,60, TimeUnit.SECONDS,new LinkedBlockingDeque<>(1000));



    public void   governanceAssess(String assessDate ){

        //1  准备待考评的表的清单 List<TableMetaInfo>
        List<TableMetaInfo> tableMetaInfoList = tableMetaInfoService.getListWithExtra(assessDate);
        System.out.println("tableMetaInfoList = " + tableMetaInfoList);

        //2  准备待考评的指标的清单  List<GovernanceMetric>
        List<GovernanceMetric> metricList = governanceMetricService.list(new QueryWrapper<GovernanceMetric>().ne("is_disabled","1"));

        //3  任务定义清单
        List<TDsTaskDefinition>  taskDefinitionList= tDsTaskDefinitionService.getTaskDefinitionList();//最新运行成功的实例的定义
        Map<String, TDsTaskDefinition> taskDefinitionMap = taskDefinitionList.stream().collect(Collectors.toMap(taskDefinition -> taskDefinition.getName(), taskDefinition -> taskDefinition));

        //4 任务的实例清单
        Map<String, List<TDsTaskInstance>>  tDsTaskInstanceMap= tDsTaskInstanceService.getTaskInstanceListMap();

        //3  双重遍历 两个清单 对每张表每个指标逐个进行考评  产生 List<GovernanceAssessDetail>
        List<GovernanceAssessDetail> assessDetailList = new ArrayList<>(tableMetaInfoList.size()* metricList.size());

        List<CompletableFuture<GovernanceAssessDetail>>  futureList=new ArrayList<>();

        long l = System.currentTimeMillis();
        for (TableMetaInfo tableMetaInfo : tableMetaInfoList) {
            for (GovernanceMetric metric : metricList){
                //判断考评表是否在指标的白名单中，如果是则直接跳过考评
                if (metric.getSkipAssessTables()!=null&&metric.getSkipAssessTables().contains(tableMetaInfo.getTableName())){
                    continue;
                }

                AssessParam assessParam= new AssessParam();
                assessParam.setAssessDate(assessDate);//1考评日期
                assessParam.setTableMetaInfo(tableMetaInfo);//2表信息
                assessParam.setGovernanceMetric(metric);//3 指标信息
                assessParam.setTableMetaInfoList(tableMetaInfoList); //4 目前所有表的集合
                assessParam.setTDsTaskDefinition(taskDefinitionMap.get(tableMetaInfo.getSchemaName()+"."+tableMetaInfo.getTableName())); //5 任务定义
                assessParam.setTDsTaskInstanceList(tDsTaskInstanceMap.get(tableMetaInfo.getSchemaName()+"."+tableMetaInfo.getTableName())); //6  任务实例清单

                CompletableFuture<GovernanceAssessDetail> future = CompletableFuture.supplyAsync(() -> {
                    GovernanceAssessDetail governanceAssessDetail = this.assessMetric(assessParam);
                    return governanceAssessDetail;
                },threadPoolExecutor);

               // GovernanceAssessDetail governanceAssessDetail = this.assessMetric( assessParam);
                //assessDetailList.add(governanceAssessDetail);
                futureList.add(future);
            }

        }
        //期货兑现货
        assessDetailList = futureList.stream().map(future -> future.join()).collect(Collectors.toList());

        //计算耗时
        System.err.println("governanceAssess 耗时 = " + (System.currentTimeMillis()-l));

        //4  保存到数据库中
        this.remove(new QueryWrapper<GovernanceAssessDetail>().eq("assess_date",assessDate));
        saveBatch(assessDetailList);

    }

    @Autowired
    SpringBeanProvider springBeanProvider ;

    private GovernanceAssessDetail assessMetric(AssessParam assessParam){




        String metricCode = assessParam.getGovernanceMetric().getMetricCode();
        //程序是否有办法根据metricCode 这个字符串来动态选择创建的考评器  1 可以利用反射
         //2 利用springboot的组件容器 来动态获得某个字符串的组件
        Assessor assessor =springBeanProvider.getBean(metricCode,Assessor.class);
        GovernanceAssessDetail governanceAssessDetail = assessor.assess(  assessParam);
        return governanceAssessDetail;
    }
}
